package com.atguigu.gmall.realtime.app.dim.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.constant.GmallConfig;
import com.atguigu.gmall.realtime.util.PhoenixUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Collection;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

public class TableProcessFunction  extends BroadcastProcessFunction<JSONObject, String, JSONObject> {


    MapStateDescriptor<String, TableProcess> tableProcessMapStateDescriptor;

    public   TableProcessFunction(MapStateDescriptor<String, TableProcess> tableProcessMapStateDescriptor){
        this.tableProcessMapStateDescriptor=tableProcessMapStateDescriptor;
    }
//    {
//        "database": "gmall",
//            "xid": 52343,
//            "data": {
//        "birthday": "2005-02-14",
//                "gender": "F",
//                "create_time": "2020-06-14 17:36:43",
//                "login_name": "834sko4sh",
//                "nick_name": "香香9999",
//                "name": "何香",
//                "user_level": "2",
//                "phone_num": "13843974462",
//                "id": 2,
//                "email": "834sko4sh@sina.com"
//    },
//        "old": {
//        "nick_name": "香香88888"
//    },
//        "commit": true,
//            "type": "update",
//            "table": "user_info",
//            "ts": 1665173630
//    }
    //  主流数据到达时 要做的处理 ：   根据广播流的配置表状态信息进行筛选
    @Override
    public void processElement(JSONObject gmallDataJson, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
      //  System.out.println("主流数据：" + gmallDataJson);
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(tableProcessMapStateDescriptor);
        //1  判断主流数据的表 是否在状态中，如果在保留  ，如果不再丢弃
        String tableName = gmallDataJson.getString("table");
        TableProcess tableProcess = broadcastState.get(tableName);
        String optType = gmallDataJson.getString("type");
        if(tableProcess!=null&&!optType.equals("bootstrap-start")&&!optType.equals("bootstrap-complete")){
            //2  如果保留，剪裁字段：根据配置表中规定的字段保留，不需要的字段删除
            //2.1  把要提取的字段放到一个set里 （便于判断）
            String sinkColumns = tableProcess.getSinkColumns();
            String[] colArr = sinkColumns.split(",");
            Set colset =new HashSet<String>();
            for (String col : colArr) {
                colset.add(col);
            }
            //2.2 把数据的键值对提取出来
            JSONObject dataJsonObj = gmallDataJson.getJSONObject("data");
            Set<Map.Entry<String, Object>> dataKeyValue = dataJsonObj.entrySet();
            //2.3 迭代键值对，如果某个字段不存在与set中，则remove掉
            dataKeyValue.removeIf(entry ->!colset.contains(entry.getKey()) );
            //3  补充要真正写入到phoenix表的表名
            gmallDataJson.put("sink_table",tableProcess.getSinkTable());


            //4 输出到out中
            out.collect(gmallDataJson);
        }




    }

    //  广播流数据到达时，要做的处理：  更新广播状态
    @Override
    public void processBroadcastElement(String tableProcessJson, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        //广播流数据 ：{"before":null,
        // "after":{"source_table":"base_category3","sink_table":"dim_base_category3","sink_columns":"id,name,category2_id","sink_pk":"id","sink_extend":null},
        // "source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1665199271947,"snapshot":"false","db":"gmall_realtime_config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1665199271947,"transaction":null}
     //   System.out.println("广播流数据 ：" + tableProcessJson);
        // 1  json转成 jsonObj
        JSONObject tableProcessJsonObj = JSON.parseObject(tableProcessJson);
        // 2   提取操作符
        String operator = tableProcessJsonObj.getString("op");

        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(tableProcessMapStateDescriptor);

        // 3  区分广播流的操作类型 ，
        if(!operator.equals("d")){ //不是删除的时候
            //  3.1  如果时新增修改+全量查询 直接维护进状态
            //    3.1.1  把after中的数据提取出来 转成javaBean  TableProcess
            TableProcess tableProcess = tableProcessJsonObj.getObject("after", TableProcess.class);
            //    3.1.2 检查phoenix是否有该维表  ，如果没有生成建表语句，执行建表
            checkTable(tableProcess);
            //     3.1.3  写入状态
            broadcastState.put(tableProcess.getSourceTable(),tableProcess);
        }else{
            //   3.2 如果是删除 要从状态中移除
            //    3.2.1  把before中的数据提取出来 转成javaBean  TableProcess
            TableProcess tableProcess = tableProcessJsonObj.getObject("before", TableProcess.class);
            // 3.2.2 根据表名删除状态
            broadcastState.remove(tableProcess.getSourceTable());
        }






        //



    }


    private void  checkTable(TableProcess tableProcess){
        // create table if not exists  source_table
        // (    col1  varchar  [primary key],  //根据sink_columns 进行拼接
        //      col2  varchar ...
        // ) [SALT_BUCKETS = 3]

        StringBuilder createTableSQL =
                new StringBuilder("create table if not exists  "+ GmallConfig.PHOENIX_SCHEMA+"."+tableProcess.getSinkTable()+"(");

        String[] sinkCols = tableProcess.getSinkColumns().split(",");
        for (int i = 0; i < sinkCols.length; i++) {
            String sinkCol = sinkCols[i];
            if(i>0){
                createTableSQL.append(", ");
            }
            createTableSQL.append(sinkCol+" varchar ");
            if(sinkCol.equals(tableProcess.getSinkPk())){
                createTableSQL.append(" primary key");
            }

        }



        createTableSQL.append(")");
        if(tableProcess.getSinkExtend()!=null&&tableProcess.getSinkExtend().trim().length()>0){
            createTableSQL.append(tableProcess.getSinkExtend());
        }
     //   System.out.println( createTableSQL);
        PhoenixUtil.executeSql(createTableSQL.toString());



    }
}
